package beautiful.butterfly.drds.data_exchange.job;


import beautiful.butterfly.drds.data_exchange.communicator.AbstractContainerCommunicator;
import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.constant.State;
import beautiful.butterfly.drds.data_exchange.error_code.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.error_code.FrameworkErrorCode;
import beautiful.butterfly.drds.data_exchange.report.Message;
import beautiful.butterfly.drds.data_exchange.report.Messages;
import beautiful.butterfly.drds.data_exchange.taskgroup.TaskGroupContainer;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.TaskGroupContainerTask;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import beautiful.butterfly.drds.data_exchange.util.ErrorRecordChecker;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Scheduler
{
    private static final Logger LOG = LoggerFactory
            .getLogger(Scheduler.class);

    private ErrorRecordChecker errorRecordChecker;

    private AbstractContainerCommunicator containerCommunicator;

    private Long jobId;
    private ExecutorService executorService;

    public Scheduler(AbstractContainerCommunicator containerCommunicator)
    {
        this.containerCommunicator = containerCommunicator;
    }

    public Long getJobId()
    {
        return jobId;
    }

    public void schedule(List<Configuration> data)
    {
        Validate.notNull(data, "scheduler配置不能为空");
        int jobReportIntervalInMillSec = data.get(0).getInt(
                Constants.container_job_reportinterval, 30000);
        int jobSleepIntervalInMillSec = data.get(0).getInt(
                Constants.container_job_sleepinterval, 10000);

        this.jobId = data.get(0).getLong(
                Constants.container_job_id);

        errorRecordChecker = new ErrorRecordChecker(data.get(0));

        /**
         * 给 taskGroupContainer 的 Message 注册
         */
        this.containerCommunicator.registerCommunication(data);

        int totalTasks = calculateTaskCount(data);
        startAllTaskGroup(data);

        Message lastJobContainerMessage = new Message();

        long lastReportTimeStamp = System.currentTimeMillis();
        try
        {
            while (true)
            {
                /**
                 * step 1: collect job stat
                 * step 2: getReport info, then report it
                 * step 3: errorRecordChecker do check
                 * step 4: dealSucceedStat();
                 * step 5: dealKillingStat();
                 * step 6: dealFailedStat();
                 * step 7: refresh last job stat, and then sleep for next while
                 *
                 * above steps, some ones should report info to DS
                 *
                 */
                Message nowJobContainerMessage = this.containerCommunicator.collect();
                nowJobContainerMessage.setTimestamp(System.currentTimeMillis());
                LOG.debug(nowJobContainerMessage.toString());

                //汇报周期
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > jobReportIntervalInMillSec)
                {
                    Message reportMessage = Messages
                            .getReportCommunication(nowJobContainerMessage, lastJobContainerMessage, totalTasks);

                    this.containerCommunicator.report(reportMessage);
                    lastReportTimeStamp = now;
                    lastJobContainerMessage = nowJobContainerMessage;
                }

                errorRecordChecker.checkRecordLimit(nowJobContainerMessage);

                if (nowJobContainerMessage.getState() == State.succeeded)
                {
                    LOG.info("Scheduler accomplished all tasks.");
                    break;
                }

                if (isJobKilling(this.getJobId()))
                {
                    dealKillingStat(this.containerCommunicator, totalTasks);
                } else if (nowJobContainerMessage.getState() == State.failed)
                {
                    dealFailedStat(this.containerCommunicator, nowJobContainerMessage.getThrowable());
                }

                Thread.sleep(jobSleepIntervalInMillSec);
            }
        } catch (InterruptedException e)
        {
            // 以 failed 状态退出
            LOG.error("捕获到InterruptedException异常!", e);

            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR, e);
        }

    }

    protected void startAllTaskGroup(List<Configuration> configurationList)
    {
        this.executorService = Executors
                .newFixedThreadPool(configurationList.size());

        for (Configuration taskGroupConfiguration : configurationList)
        {
            TaskGroupContainerTask taskGroupContainerTask = newTaskGroupContainerRunner(taskGroupConfiguration);
            this.executorService.execute(taskGroupContainerTask);
        }

        this.executorService.shutdown();
    }

    private TaskGroupContainerTask newTaskGroupContainerRunner(
            Configuration configuration)
    {
        TaskGroupContainer taskGroupContainer = new TaskGroupContainer(configuration);

        return new TaskGroupContainerTask(taskGroupContainer);
    }

    protected void dealFailedStat(AbstractContainerCommunicator frameworkCollector, Throwable throwable)
    {
        this.executorService.shutdownNow();
        throw DataExchangeException.asDataExchangeException(
                FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, throwable);
    }

    protected void dealKillingStat(AbstractContainerCommunicator frameworkCollector, int totalTasks)
    {
        //通过进程退出返回码标示状态
        this.executorService.shutdownNow();
        throw DataExchangeException.asDataExchangeException(FrameworkErrorCode.KILLED_EXIT_VALUE,
                "job killed status");
    }

    private int calculateTaskCount(List<Configuration> data)
    {
        int totalTasks = 0;
        for (Configuration taskGroupConfiguration : data)
        {
            totalTasks += taskGroupConfiguration.getListConfiguration(
                    Constants.task_list).size();
        }
        return totalTasks;
    }


    protected boolean isJobKilling(Long jobId)
    {
        return false;
    }
}
